package cn._51doit.day10;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @create: 2021-10-28 14:26
 * @author: 今晚打脑斧先森
 * @program: Demo01
 * @Description:
 *      练习左外连接
 *      数据1：9999,o1,苹果,5
 *      数据2：9999,o1,5
 **/
public class Demo01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        //第一个流
        DataStreamSource<String> lines1 = env.socketTextStream("doit01", 8888);
        SingleOutputStreamOperator<Tuple4<Long,String, String, Integer>> tpStream1 = lines1.map(new MapFunction<String, Tuple4<Long,String, String, Integer>>() {
            @Override
            public Tuple4<Long,String, String, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                //时间戳，订单，商品名称，单价
                return Tuple4.of(Long.parseLong(split[0]),split[1], split[2], Integer.parseInt(split[3]));
            }
        }).setParallelism(1);
        SingleOutputStreamOperator<Tuple4<Long, String, String, Integer>> tpAndWaterMark = tpStream1.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple4<Long, String, String, Integer>>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Tuple4<Long, String, String, Integer> element) {
                return element.f0;
            }
        });
        //第二个流
        DataStreamSource<String> lines2 = env.socketTextStream("doit01", 9999);
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> tpStream2 = lines2.map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
            @Override
            public Tuple3<Long, String, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                //时间戳，商品订单，个数
                return Tuple3.of(Long.parseLong(split[0]), split[1], Integer.parseInt(split[2]));
            }
        }).setParallelism(1);
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> tpAndWaterMark2 = tpStream2.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Long, String, Integer>>(Time.seconds(0)) {
            @Override
            public long extractTimestamp(Tuple3<Long, String, Integer> element) {
                return element.f0;
            }
        });
        DataStream<Tuple5<String, String, Integer, Integer, Integer>> res = tpAndWaterMark.coGroup(tpAndWaterMark2)
                .where(t1 -> t1.f1)
                .equalTo(t2 -> t2.f1)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                //时间戳，订单，商品名称，单价   //时间戳，商品订单，个数  //订单，商品名称，单价 ，个数，总价
                .apply(new CoGroupFunction<Tuple4<Long, String, String, Integer>, Tuple3<Long, String, Integer>, Tuple5<String, String, Integer, Integer, Integer>>() {
                    @Override
                    public void coGroup(Iterable<Tuple4<Long, String, String, Integer>> first, Iterable<Tuple3<Long, String, Integer>> second, Collector<Tuple5<String, String, Integer, Integer, Integer>> out) throws Exception {
                        for (Tuple4<Long, String, String, Integer> left : first) {
                            boolean flag = false;
                            for (Tuple3<Long, String, Integer> right : second) {
                                flag = true;
                                String order = left.f1;
                                String name = left.f2;
                                Integer price = left.f3;
                                Integer count = right.f2;
                                Integer sumPrice = price * count;
                                out.collect(Tuple5.of(order, name, price, count, sumPrice));
                            }
                            if (!flag){
                                out.collect(Tuple5.of(left.f1, left.f2,  left.f3, null, null));
                            }
                        }
                    }
                });
        res.addSink(new SinkFunction<Tuple5<String, String, Integer, Integer, Integer>>() {
            @Override
            public void invoke(Tuple5<String, String, Integer, Integer, Integer> value) throws Exception {
                System.out.println("订单："+value.f0+" 商品名称："+value.f1+" 单价："+value.f2+" 购买个数："+value.f3+" 总价："+value.f4);
            }
        });
        env.execute();
    }
}
